<?php

$server = new swoole_websocket_server("0.0.0.0", 8899);
if (PHP_OS == 'Darwin'){
    define('PG_HOST', 'localhost');
    define('PG_PORT', 5432);
    define('PG_DBNAME', 'dftg');
    define('PG_USER', 'dftg');
    define('PG_PASSWORD', 'dftg');
}
else{
    define('PG_HOST', 'rm-bp10b898828yjw615.pg.rds.aliyuncs.com');
    define('PG_PORT', 3433);
    define('PG_DBNAME', 'dftg');
    define('PG_USER', 'odoo');
    define('PG_PASSWORD', 'Dftgodoopwd800');
}


$server->set(array(
    'worker_num' => 2,    //worker process num
    'task_worker_num' => 2, //任务进程
    'backlog' => 128,   //listen backlog
    'max_request' => 500,
    'dispatch_mode' => 1,
    'open_tcp_keepalive' => 1,
    'tcp_keepidle' => 60,
    'tcp_keepcount' => 1,
    'tcp_keepinterval' => 30,
    //心跳检测
    'heartbeat_check_interval' => 30,
    'heartbeat_idle_time' => 600,
));

//状态共享表
$server->table = new swoole_table(1);
$server->table->column('is_start_listen_purchase', swoole_table::TYPE_INT, 4);
$server->table->column('is_start_listen_move', swoole_table::TYPE_INT, 4);
$server->table->column('listen_move_client_fd', swoole_table::TYPE_INT, 8);
$server->table->create();
$server->table->set('pg_listen', array('is_start_listen_purchase' => 0, 'is_start_listen_move' => 0, 'listen_move_client_fd' => 0));


//客户端id类型共享表
$server->fd_table = new swoole_table(100);
$server->fd_table->column('type', swoole_table::TYPE_INT, 4);
$server->fd_table->create();

$server->on('open', function (swoole_websocket_server $server, $request) {
    echo date('Y-m-d H:i:s') . " server worker_id:{$server->worker_id}: handshake success with fd{$request->fd}\n";
    $table_row = $server->table->get('pg_listen');
    if ($table_row['is_start_listen_purchase'] == 0){
        $server->table->set('pg_listen', array('is_start_listen_purchase' => 1));
        $task_id = $server->task("start listen purchase");
    }
});

$server->on('message', function (swoole_websocket_server $server, $frame) {
	if ($frame->data == 'get data'){
	    $server->fd_table->set($frame->fd, array('type' => 1));
		$db = new PDO(
		    "pgsql:dbname=" . PG_DBNAME . " host=" . PG_HOST . " port=" . PG_PORT, PG_USER, PG_PASSWORD, [
		        PDO::ATTR_ERRMODE            => PDO::ERRMODE_EXCEPTION,
		        PDO::ATTR_DEFAULT_FETCH_MODE => PDO::FETCH_ASSOC,
		    ]
		);
		if ($db){
			$res = array();
			$ps = $db->query("SELECT * FROM opportunity_stat_json");
			$row = $ps->fetch();
			$res['opportunity_changed'] = json_decode($row['json_agg']);
			$ps = $db->query("SELECT * FROM so_line_change_stat_json");
			$row = $ps->fetch();
			$res['sale_order_line_change_changed'] = json_decode($row['json_agg']);
		    $server->push($frame->fd, json_encode($res));
		    $db = null;
		}
	}
	if ($frame->data == 'listen move'){
	    $server->fd_table->set($frame->fd, array('type' => 2));
        $table_row = $server->table->get('pg_listen');
        $table_row['listen_move_client_fd'] = $frame->fd;
        if ($table_row['is_start_listen_move'] == 0){
            $server->table->set('pg_listen', array('is_start_listen_move' => 1));
            $task_id = $server->task("start listen move");
        }
	}
    echo date('Y-m-d H:i:s') . " receive from {$frame->fd}:{$frame->data},opcode:{$frame->opcode},fin:{$frame->finish}\n";
});

$server->on('close', function ($server, $fd) {
    if ($server->fd_table->exist($fd)){
        $server->fd_table->del($fd);
    }
    echo date('Y-m-d H:i:s') . " client {$fd} closed\n";
});

$server->on('workerStart', function($server, $worker_id){
	echo date('Y-m-d H:i:s') ." WorkerStart {$worker_id}\n";
});

$server->on('managerStart', function($server){
	echo "managerStart\n";
});


$server->on('task', function($server, $task_id, $from_id, $data){
    echo date('Y-m-d H:i:s') . " new $data task={$task_id}\n";
    $db = new PDO(
        "pgsql:dbname=" . PG_DBNAME . " host=" . PG_HOST . " port=" . PG_PORT, PG_USER, PG_PASSWORD, [
            PDO::ATTR_ERRMODE            => PDO::ERRMODE_EXCEPTION,
            PDO::ATTR_DEFAULT_FETCH_MODE => PDO::FETCH_ASSOC,
            PDO::ATTR_PERSISTENT => true
        ]
    );
    $table_row = $server->table->get('pg_listen');
    //仓库客户端id
    $stock_move_client_fd = $table_row['listen_move_client_fd'];
    //采购获取监控数据
    if ($data == 'start listen purchase'){
        $db->exec('LISTEN opportunity_changed;LISTEN sale_order_line_change_changed;');
        $start_time = time();
        $diff_time = 0;
        while($diff_time < 300){//超过5分钟重启任务，重启任务的间隙可能漏掉数据变化
            $result = $db->pgsqlGetNotify(PDO::FETCH_ASSOC, 60000);
            if ($result){
                $res = array();
                $res[$result['message']] = json_decode($result['payload']);
                foreach($server->connections as $fd){
                    $fd_type_row = $server->fd_table->get($fd);
                    if ($fd_type_row == false || $fd_type_row['type'] == 1){
                        echo date('Y-m-d H:i:s') . " push to fd={$fd}\n";
                        $server->push($fd, json_encode($res));
                    }
                }
            }
            else{
                $fd_count = count($server->connections);
                echo date('Y-m-d H:i:s') . " fd count={$fd_count}\n";
            }
            $diff_time = time() -  $start_time;
        }
    }
    //仓库获取监控数据
    if ($data == 'start listen move'){
        $db->exec('LISTEN stock_move_done');
        $start_time = time();
        $diff_time = 0;
        while($diff_time < 300){//超过5分钟重启任务，重启任务的间隙可能漏掉数据变化
            $result = $db->pgsqlGetNotify(PDO::FETCH_ASSOC, 60000);
            if ($result){
                $res = array();
                $res[$result['message']] = json_decode($result['payload']);
                foreach($server->connections as $fd){
                    $fd_type_row = $server->fd_table->get($fd);
                    if ($fd_type_row == false || $fd_type_row['type'] == 2){
                        echo date('Y-m-d H:i:s') . " push to fd={$fd}\n";
                        $server->push($fd, json_encode($res));
                    }
                }
            }
            else{
                $fd_count = count($server->connections);
                echo date('Y-m-d H:i:s') . " fd count={$fd_count}\n";
            }
            $diff_time = time() -  $start_time;
        }
    }
	return true;
});
$server->on('finish', function($server, $task_id, $data){
    echo date('Y-m-d H:i:s') . " finish task={$task_id}\n";
    if ($data == 'start listen purchase'){
        $task_id = $server->task("start listen purchase");
    }
    if ($data == 'start listen move'){
        $task_id = $server->task("start listen move");
    }
});



$server->start();


/** u 移动入库触发器
DROP TRIGGER move_stock_in_trigger ON stock_move;

CREATE TRIGGER move_stock_in_trigger
 after UPDATE
 ON stock_move FOR EACH ROW
 WHEN (OLD.state='assigned' AND NEW.state='done' AND NEW.picking_type_id=1)
 EXECUTE PROCEDURE notify_move_done();



CREATE OR REPLACE FUNCTION notify_move_done() RETURNS TRIGGER AS $$
BEGIN
    -- SELECT name, product_qty FROM stock_move WHERE
    PERFORM pg_notify('stock_move_done', row_to_json(NEW)::text);
    RETURN NULL;
END;
$$ LANGUAGE plpgsql;
*/